package appcore

import (
	"sync"

	"gitlab.upchinaproduct.com/taf/tafgo/taf"
)

var (
	once      sync.Once
	instance  *WorkerPool
	loggerObj = taf.GetDayLogger("WorkerPool", 1)
)

type Task struct {
	Id int
	// 配置文件中的数据库id
	DatabaseID string
	// 查出来源数据的键值对 key为数据库列 value为数据库值
	DataBaseKV map[string]interface{}
}

type WorkerPool struct {
	taskQueue   chan Task
	workerCount int
	done        chan bool
}

func GetWorkerPoolInstance() *WorkerPool {
	once.Do(func() {
		instance = NewWorkerPool(8)
	})
	return instance
}

func NewWorkerPool(workerCount int) *WorkerPool {
	return &WorkerPool{
		taskQueue:   make(chan Task, 8000), // buffered channel for task queue
		workerCount: workerCount,
		done:        make(chan bool),
	}
}

func (wp *WorkerPool) Start() {
	for i := 0; i < wp.workerCount; i++ {
		go wp.worker(i)
	}
}

func (wp *WorkerPool) worker(id int) {
	for task := range wp.taskQueue {
		// grep "warning HandleDestinationDatabases" CRM.DataConvertServer_WorkerPool_*.log
		loggerObj.Debugf("Worker %d processing task %d for DatabaseID %s", id, task.Id, task.DatabaseID)
		var retSrc, errHandleSource = HandleDestinationDatabases(loggerObj, task)
		if retSrc != 0 || errHandleSource != nil {
			loggerObj.Debugf("{warning HandleDestinationDatabases task^ret^error}|%+v|%d|%+v", task, retSrc, errHandleSource)
		} else {
			loggerObj.Debugf("{success HandleDestinationDatabases task^ret^error}|%+v|%d|%+v", task, retSrc, errHandleSource)
		}
	}
}

func (wp *WorkerPool) AddTask(task Task) {
	wp.taskQueue <- task
}

func (wp *WorkerPool) Stop() {
	close(wp.taskQueue)
	wp.done <- true
}

func (wp *WorkerPool) Wait() {
	<-wp.done
}
